package modbus_master_connector

import (
	"encoding/binary"
	"errors"
	"fmt"
	"github.com/sirupsen/logrus"
	"go-iot-gateway/internal/apk/maths"
	"go-iot-gateway/internal/channel"
	modbus_connector "go-iot-gateway/internal/channel/connector/modbus"
	channel_define "go-iot-gateway/internal/channel/define"
	"go-iot-gateway/internal/conf_table"
	"go-iot-gateway/internal/conf_table/define"
	"go-iot-gateway/internal/storage"
	storage_define "go-iot-gateway/internal/storage/define"
	"go-iot-gateway/pkg/modbus"
	"sort"
	"strconv"
	"sync"
	"time"
)

// modbus 实现类
type innerConnector interface {
	_init(protocolConfInfos []*conf_table_define.ProtocolInfo)
	_connect() error
	_close()
	_numOfRegisterPerFrame() uint16
	_sendLock()
	_sendUnLock()
	_writeRetryOnFail() uint8
}

// modbus 连接通道 https://pkg.go.dev/github.com/goburrow/modbus#section-readme
type abstractModbusConnector struct {
	ChannelInfo       channel_define.ChannelInfo
	modbusClient      modbus.Client
	isClosed          bool
	channelManager    *channel.ChannelManager
	readTagGroupByDev map[string][]*conf_table_define.TagInfo //读点位， key =Dev_id
	//writeTagList      []*conf_table_define.TagInfo            //写点位
	lock          sync.Mutex
	_connector    innerConnector
	numberOfError int64 // 发生错误的次数
	logger        *logrus.Logger
}

// {ED3C45F3-7CDB-4050-90BF-3EC88225273F} MODBUS_TCP
func (m *abstractModbusConnector) Open() {
	//todo 只做一次

	m.isClosed = false

	tagConfInfos := conf_table.FindAllTagByChlId(m.ChannelInfo.ChlId)

	if len(tagConfInfos) > 0 {
		m.tidyConfTable(m.ChannelInfo.ChlId, tagConfInfos)
	}

	go func() { m.connect() }()

	if len(tagConfInfos) <= 0 {
		return
	}

	//todo 如果是 tcp 可根据设备 并发读
	go func() {
		m.pull()
	}()

	go func() {
		m.healCheck()
	}()

	//todo 定时任务将过久没有上报的tag 标识为坏点
}

func (m *abstractModbusConnector) Info() channel_define.ChannelInfo {
	return m.ChannelInfo
}

func (m *abstractModbusConnector) ReadVal(key string) storage_define.ValObject {
	//todo 读数据
	return nil
}

func (m *abstractModbusConnector) WriteVal(key string, val interface{}) {
	//todo 写失败重试
	// todo modbus 广播
	//TODO implement me
	//results, err = client.WriteMultipleRegisters(1, 2, []byte{0, 3, 0, 4})
	//fmt.Printf(string(results))
	//results, err = client.WriteMultipleCoils(5, 10, []byte{4, 3})
	//fmt.Printf(string(results))
	//logger.Connector().Error(err)
	panic("implement me")
}

func (m *abstractModbusConnector) increaseNumberOfError() {
	m.lock.Lock()
	defer m.lock.Unlock()

	m.numberOfError = m.numberOfError + 1
}

func (m *abstractModbusConnector) clearNumberOfError() {
	m.lock.Lock()
	defer m.lock.Unlock()
	//todo 优化使用 原子操作
	m.numberOfError = 0
}

// 写值
//func (m *modbusConnector) writeVal(key string, val interface{}) {
//
//}

// 健康监测
func (m *abstractModbusConnector) healCheck() {
	for {
		maxErrorNum := int64(32)
		if m.numberOfError >= maxErrorNum {
			m.reconnect(channel_define.CONNECT_STATUS_ERROR, "错误次数达到阈值") //如果错误达到 32 处理, 就进行重连
		}
		time.Sleep(7 * time.Second) //每7 秒执行一次
	}
}

func (m *abstractModbusConnector) Close() {
	m.isClosed = true
	m.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CLOSED, "主动关闭")
	m._connector._close()
}

// 整理点表
func (m *abstractModbusConnector) tidyConfTable(chlId string, tagConfInfos []*conf_table_define.TagInfo) {

	//读点表按设备分组,初始化 m.readTagGroupByDev
	deviceInfos := conf_table.FindAllDeviceByChlId(chlId)
	m.readTagGroupByDev = make(map[string][]*conf_table_define.TagInfo, len(deviceInfos))
	for _, dev := range deviceInfos {
		if !dev.IsValid() {
			continue
		}
		//todo 设备的 通讯地址验证， 必须为整型
		m.readTagGroupByDev[dev.Dev_id] = make([]*conf_table_define.TagInfo, 0, 125)
	}
	//m.writeTagList = make([]*conf_table_define.TagInfo, 0, 32)

	for _, tag := range tagConfInfos {
		//过滤无效的设备
		if !tag.DeviceInfo.IsValid() {
			continue
		}

		//过滤不是整数地址的
		_, err := strconv.ParseInt(tag.Obj_addr, 10, 32)
		if err != nil {
			continue
		}

		//过滤可读的点位
		if !tag.CanRead() {
			continue
		}

		switch tag.Collect_type {
		case modbus_connector.COILS: //读写离散输出区
			if tag.Tag_type == conf_table_define.TAG_TYPE_YX || tag.Tag_type == conf_table_define.TAG_TYPE_YC || tag.Tag_type == conf_table_define.TAG_TYPE_YM {
				m.readTagGroupByDev[tag.DeviceInfo.Dev_id] = append(m.readTagGroupByDev[tag.DeviceInfo.Dev_id], tag)
			} else {
				//m.writeTagList = append(m.writeTagList, tag)
			}
		case modbus_connector.DISCRETE_INPUTS: //读离散输入区
			m.readTagGroupByDev[tag.DeviceInfo.Dev_id] = append(m.readTagGroupByDev[tag.DeviceInfo.Dev_id], tag)
		case modbus_connector.HOLDING_REGISTERS, modbus_connector.HOLDING_MUTIL_REGISTERS: //读写保持寄存器
			if tag.Tag_type == conf_table_define.TAG_TYPE_YX || tag.Tag_type == conf_table_define.TAG_TYPE_YC || tag.Tag_type == conf_table_define.TAG_TYPE_YM {
				m.readTagGroupByDev[tag.DeviceInfo.Dev_id] = append(m.readTagGroupByDev[tag.DeviceInfo.Dev_id], tag)
			} else {
				//m.writeTagList = append(m.writeTagList, tag)
			}
		case modbus_connector.INPUT_REGISTERS: //读输入寄存器
			m.readTagGroupByDev[tag.DeviceInfo.Dev_id] = append(m.readTagGroupByDev[tag.DeviceInfo.Dev_id], tag)
		default:
			continue
		}

	}

	//按地址排序,可以进行批量采集
	for _, dev := range deviceInfos {
		if !dev.IsValid() {
			continue
		}
		tagInfos := m.readTagGroupByDev[dev.Dev_id]
		sort.Slice(tagInfos, func(i, j int) bool {
			return tagInfos[i].Obj_addr_int < tagInfos[j].Obj_addr_int
		})
	}

}

// 按采集类型分组
func groupByCollectType(tagInfos []*conf_table_define.TagInfo) ([]*conf_table_define.TagInfo, []*conf_table_define.TagInfo, []*conf_table_define.TagInfo, []*conf_table_define.TagInfo) {
	coils := make([]*conf_table_define.TagInfo, 0, 20)
	discreteInputs := make([]*conf_table_define.TagInfo, 0, 20)
	holdingRegisters := make([]*conf_table_define.TagInfo, 0, 125)
	inputRegisters := make([]*conf_table_define.TagInfo, 0, 20)

	for _, tag := range tagInfos {
		switch tag.Collect_type {
		case modbus_connector.COILS:
			coils = append(coils, tag)
		case modbus_connector.DISCRETE_INPUTS:
			discreteInputs = append(discreteInputs, tag)
		case modbus_connector.HOLDING_REGISTERS, modbus_connector.HOLDING_MUTIL_REGISTERS:
			holdingRegisters = append(holdingRegisters, tag)
		case modbus_connector.INPUT_REGISTERS:
			inputRegisters = append(inputRegisters, tag)
		}
	}

	return coils, discreteInputs, holdingRegisters, inputRegisters
}

func (m *abstractModbusConnector) reconnect(status channel_define.ConnectStatusType, cause string) {
	m.ChannelInfo.SetStatus(status, cause)
	m._connector._close()
	m.clearNumberOfError()
	go func() { m.connect() }()
}

func (m *abstractModbusConnector) connect() {
	if m.ChannelInfo.Status == channel_define.CONNECT_STATUS_CONNECTING || m.ChannelInfo.Status == channel_define.CONNECT_STATUS_CONNECTED {
		return
	}
	m.lock.Lock()
	if m.ChannelInfo.Status == channel_define.CONNECT_STATUS_CONNECTING || m.ChannelInfo.Status == channel_define.CONNECT_STATUS_CONNECTED {
		return
	}
	m.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTING, "连接中")
	m.lock.Unlock()

	for {
		err := m._connector._connect()
		if err != nil {
			m.logger.Warnf("%s通道连接失败 %+v %s", m.ChannelInfo.Name, m.ChannelInfo.ConnectParams, err)
			m.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_ERROR, "连接错误 "+err.Error())
			time.Sleep(4 * time.Second)
			continue
		}
		m.logger.Debugf("%s通道连接成功  ", m.ChannelInfo.Name)
		m.clearNumberOfError()
		m.ChannelInfo.SetStatus(channel_define.CONNECT_STATUS_CONNECTED, "连接成功")
		break
	}
}

// 获取数据
func (m *abstractModbusConnector) pull() {
	//交互频率
	frequency := m.ChannelInfo.Frequency

	for !m.isClosed {

		beginTime := time.Now().UnixMilli()

		for devId, tagInfos := range m.readTagGroupByDev {

			if m.isClosed {
				return
			}

			for m.ChannelInfo.Status != channel_define.CONNECT_STATUS_CONNECTED && !m.isClosed {
				time.Sleep(time.Duration(1) * time.Second)
				continue
			}

			c1, c2, c3, c4 := groupByCollectType(tagInfos)

			m.batchRead(devId, modbus_connector.COILS, c1)
			m.batchRead(devId, modbus_connector.DISCRETE_INPUTS, c2)
			m.batchRead(devId, modbus_connector.HOLDING_REGISTERS, c3)
			m.batchRead(devId, modbus_connector.INPUT_REGISTERS, c4)

		}

		//交互频率
		consumeTime := time.Now().UnixMilli() - beginTime
		pauseTime := int64(frequency) - consumeTime
		if pauseTime > 0 && !m.isClosed {
			time.Sleep(time.Duration(pauseTime) * time.Millisecond)
		}

	}
}

// 批量读 对于Modbus RTU，一个完整的请求或响应不能超过 256字节
func (m *abstractModbusConnector) batchRead(devId string, collType int, tagInfos []*conf_table_define.TagInfo) {

	defer func() {
		if rerr := recover(); rerr != nil {
			m.logger.Errorf("[panic] batchRead error %s \n", rerr)
		}
	}()

	if tagInfos == nil || len(tagInfos) <= 0 {
		return
	}

	var err error
	maxQuantity := m._connector._numOfRegisterPerFrame()      //最大读取寄存器的数量
	slaveId := byte(tagInfos[0].DeviceInfo.Dev_comm_addr_int) //从站地址
	tagTotal := len(tagInfos)                                 //该批总的读取量

	onceBatchTag := make([]*conf_table_define.TagInfo, 0, tagTotal/4) //某批次的点位
	startAddr := uint16(0)                                            //读取的起始地址
	readQuantity := uint16(0)                                         //要读取的寄存器个数
	results := make([]storage_define.ValObject, 0, tagTotal)          //

	//todo 自动分组， 与手动分组
	//todo 如果有指定设备时间那么ts 应该是设备时间 ， 如果有读取设备时间， 要记录设备时间与当前时间的偏移值，而后求出每个ts的值
	for i, tag := range tagInfos {
		//第一个
		if readQuantity <= 0 {
			startAddr = uint16(tag.Obj_addr_int)
			readQuantity = modbus_connector.GetQuantity(tag.DataTypeStr())
			onceBatchTag = append(onceBatchTag, tag)
		} else if readQuantity > 0 {
			tagQuantity := modbus_connector.GetQuantity(tag.DataTypeStr())             //下一次的寄存器数
			isOneBatch := uint16(tag.Obj_addr_int)+tagQuantity-startAddr < maxQuantity //是否是同一批次
			if isOneBatch {
				onceBatchTag = append(onceBatchTag, tag)
				readQuantity = uint16(tag.Obj_addr_int) + tagQuantity - startAddr
			} else {
				var vals []storage_define.ValObject
				//结束旧的一批
				vals, err = m.onceBatchRead(slaveId, collType, startAddr, readQuantity, onceBatchTag)
				if err == nil {
					results = append(results, vals...)
				}

				//开始新的一批
				startAddr = uint16(tag.Obj_addr_int)
				readQuantity = modbus_connector.GetQuantity(tag.DataTypeStr())
				onceBatchTag = onceBatchTag[:0] //清空
				onceBatchTag = append(onceBatchTag, tag)
			}
		}

		//已经是最后一个了
		if i+1 >= tagTotal {
			var vals []storage_define.ValObject
			vals, err = m.onceBatchRead(slaveId, collType, startAddr, readQuantity, onceBatchTag)
			readQuantity = 0
			if err == nil {
				results = append(results, vals...)
			}
		}
	}

	//有值就往库里发
	if len(results) > 0 {
		for _, valObject := range results {
			if valObject == nil {
				continue
			}
			//todo 封装成 批对象
			storage.SaveAndNotify(valObject)
		}
	}

	//可能网络问题
	if err != nil {
		m.logger.Error("modbus 读取数据失败 ", err)
	}

}

// 一小批次的读取
func (m *abstractModbusConnector) onceBatchRead(slaveId byte, collType int, startAddr uint16, quantity uint16, tags []*conf_table_define.TagInfo) ([]storage_define.ValObject, error) {

	bytes, err := m.readModbusData(slaveId, collType, startAddr, quantity)
	if err != nil || bytes == nil {
		return nil, err
	}

	results := make([]storage_define.ValObject, 0, len(tags))
	beginAddr := tags[0].Obj_addr_int

	for _, tag := range tags {
		valObject := storage_define.NewValObject(tag)
		truBytes, err := m.truncateBytes(beginAddr, tag, bytes)

		if err == nil {
			err = valObject.SetBytes(truBytes, tag.DataFormatStr())
		}

		if err != nil {
			m.logger.Error("创建 ValObject 失败", err)
			valObject = storage_define.NewNilVal(tag).SetQos(storage_define.QOS_BAD)
		}
		results = append(results, valObject)
	}

	return results, nil
}

func (m *abstractModbusConnector) readModbusData(slaveId byte, collType int, startAddr uint16, quantity uint16) ([]byte, error) {

	defer func() {
		if rerr := recover(); rerr != nil {
			m._connector._sendUnLock()
			m.logger.Panicf("[panic] readModbusData error %s \n", rerr)
		}
	}()

	//todo 读取超时
	//每帧间隔
	waitTime := time.Duration(m.ChannelInfo.WaitTime)

	var err error
	var bytes []byte

	for i := 0; i < 1; i++ { //todo 读失败 重试 3 次 , 需要动态的来配置
		m._connector._sendLock()
		//defer m._connector._sendUnLock()
		switch collType {
		case modbus_connector.COILS: // 1 0x01,0x05号命令(读写离散输出区)  //ReadCoils 读取线圈 , 写单个线圈
			bytes, err = m.modbusClient.ReadCoils(slaveId, startAddr, quantity)
		case modbus_connector.DISCRETE_INPUTS: // 2 0x02号命令(读离散输入区) 		  //DiscreteInputs
			bytes, err = m.modbusClient.ReadDiscreteInputs(slaveId, startAddr, quantity)
		case modbus_connector.HOLDING_REGISTERS, modbus_connector.HOLDING_MUTIL_REGISTERS: // 3 0x03,0x06号命令(读写保持寄存器)  //HoldingRegisters
			bytes, err = m.modbusClient.ReadHoldingRegisters(slaveId, startAddr, quantity)
		case modbus_connector.INPUT_REGISTERS: // 4 0x04号命令(读输入寄存器)		  //InputRegisters
			bytes, err = m.modbusClient.ReadInputRegisters(slaveId, startAddr, quantity)
		default:
			err = fmt.Errorf("modbus 未知的读类型 %s", collType)
		}
		m._connector._sendUnLock()

		if err == nil {
			if waitTime > 0 {
				time.Sleep(waitTime * time.Millisecond) //每帧间隔
			}
			m.clearNumberOfError()
			return bytes, err
		}

		m.increaseNumberOfError() //添加错误次数
	}

	if err != nil {
		return nil, err
	}

	return bytes, err
}

// 根据起始地址，与当前地址计算出 byte
func (m *abstractModbusConnector) truncateBytes(beginAddr int64, tag *conf_table_define.TagInfo, bytes []byte) ([]byte, error) {
	readByteSize := modbus_connector.GetQuantity(tag.DataTypeStr()) * 2 //一个寄存器2个字节

	if beginAddr == tag.Obj_addr_int {
		return bytes[:readByteSize], nil
	}
	startByteSize := (tag.Obj_addr_int - beginAddr) * 2
	endByteSize := (startByteSize + int64(readByteSize))
	maxSize := int64(len(bytes))
	if endByteSize > maxSize {
		m.logger.Errorf("批量读取下标越界 读取开始地址 %d  结束地址 %d 最大地址 %d ", startByteSize, endByteSize, maxSize)
		return nil, errors.New("下标越界")
	}
	return bytes[startByteSize:endByteSize], nil
}

func (m *abstractModbusConnector) WriteBytes(tagId string, bytes []byte) error {

	defer func() {
		if rerr := recover(); rerr != nil {
			m._connector._sendUnLock()
			m.logger.Panicf("[panic] WriteBytes error %s \n", rerr)
		}
	}()

	if m.isClosed {
		return fmt.Errorf("通道已经关闭了")
	}

	if m.ChannelInfo.Status != channel_define.CONNECT_STATUS_CONNECTED {
		return fmt.Errorf("通道状态异常 %s", m.ChannelInfo.StatusDesc)
	}

	tagInfo, _ := conf_table.FindAllTagByTagId(tagId)
	if tagInfo == nil {
		return fmt.Errorf("配置信息丢失")
	}

	if !tagInfo.CanWrite() {
		return fmt.Errorf("写操作被关闭")
	}

	if !tagInfo.IsIntAddr() {
		return fmt.Errorf("设备配置的写地址无效 %s", tagInfo.Obj_addr)
	}

	slaveId := byte(tagInfo.DeviceInfo.Dev_comm_addr_int)

	writeVal := uint64(0)
	byteLen := len(bytes)
	switch byteLen {
	case 1:
		writeVal = uint64(bytes[0])
	case 2:
		writeVal = uint64(binary.BigEndian.Uint16(bytes))
	case 4:
		writeVal = uint64(binary.BigEndian.Uint32(bytes))
	case 8:
		writeVal = binary.BigEndian.Uint64(bytes)
	default:
		return fmt.Errorf("写报文长度无效 %d", byteLen)
	}

	var responseBytes []byte
	var err error

	retryOnFail := maths.MaxInt(int(m._connector._writeRetryOnFail()), 1) //失败重次
	quantity := modbus_connector.GetQuantity(tagInfo.DataTypeStr())
	_bytes := make([]byte, 8)
	binary.BigEndian.PutUint64(_bytes, writeVal)
	_bytes = _bytes[8-(quantity*2):]

	m._connector._sendLock()
	for i := 0; i < retryOnFail; i++ {

		m.logger.Infof("[写开始- %d] 设备 %s name %s  desc  %s \n", i, tagInfo.DeviceInfo.Dev_name, tagInfo.Tag_name, tagInfo.Tag_desc)
		switch tagInfo.Collect_type {
		case modbus_connector.DISCRETE_INPUTS: // 2 0x02号命令(读离散输入区) 		  //DiscreteInputs
			err = fmt.Errorf("modbus 0x02号命令不能进行写操作")
		case modbus_connector.INPUT_REGISTERS: // 4 0x04号命令(读输入寄存器)		  //InputRegisters
			err = fmt.Errorf("modbus 0x04号命令不能进行写操作")
		case modbus_connector.COILS: // 1 0x01,0x05号命令(读写离散输出区)  //ReadCoils 读取线圈 , 写单个线圈 (0x0F)  (0x05)
			responseBytes, err = m.modbusClient.WriteSingleCoil(slaveId, uint16(tagInfo.Obj_addr_int), uint16(writeVal))
			//todo 写多寄存器
			//responseBytes, err = m.modbusClient.WriteMultipleCoils(slaveId, uint16(tagInfo.Obj_addr_int), quantity, _bytes)
		case modbus_connector.HOLDING_REGISTERS: // 3 0x03,0x06号命令(读写保持寄存器)  //HoldingRegisters  (0x06)
			responseBytes, err = m.modbusClient.WriteSingleRegister(slaveId, uint16(tagInfo.Obj_addr_int), uint16(writeVal))
			//todo 写多寄存器
			//responseBytes, err = m.modbusClient.WriteMultipleRegisters(slaveId, uint16(tagInfo.Obj_addr_int), quantity, _bytes)
		case modbus_connector.HOLDING_MUTIL_REGISTERS: // 16 0x10号命令(读写多寄存器) (0x10)
			responseBytes, err = m.modbusClient.WriteMultipleRegisters(slaveId, uint16(tagInfo.Obj_addr_int), quantity, _bytes)
		default:
			err = fmt.Errorf("modbus 未知的读类型 %d", tagInfo.Collect_type)
		}

		if err == nil {
			break
		}

	}

	m._connector._sendUnLock()

	if err != nil {
		return err
	}

	m.logger.Infof("[写回复] % x \n", responseBytes)

	return err

}
